package com.hulk.rpc.transport.client;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ArrayUtil;
import com.google.common.collect.Maps;
import com.hulk.rpc.annotation.HulkService;
import com.hulk.rpc.common.HulkConnectService;
import com.hulk.rpc.common.RemoteContext;
import com.hulk.rpc.config.client.AppConfig;
import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.filter.ClientFilter;
import com.hulk.rpc.invoke.Invoker;
import com.hulk.rpc.invoke.InvokerUtils;
import com.hulk.rpc.loadbalance.LoadBalance;
import com.hulk.rpc.loadbalance.Weightable;
import com.hulk.rpc.param.MethodParam;
import com.hulk.rpc.remote.RemoteException;
import com.hulk.rpc.config.server.ProtocolType;
import com.hulk.rpc.utils.concurrent.ConcurrentArrayList;
import io.netty.channel.EventLoopGroup;
import io.netty.util.collection.IntObjectHashMap;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * @author xuxukang
 * @date 2019-04-09 22:05
 */
@Slf4j
@EqualsAndHashCode
public class App implements Closeable {

  private static final int max_connector_select_times = 5;
  private static final long rescue_period = TimeUnit.SECONDS.toMillis(5);
  private static final long heartbeat_period = TimeUnit.SECONDS.toMillis(5);

  // 并发的做一些建立连接、心跳等后台工作，线程数量用配置的方式更合理一些
  private static final ForkJoinPool appForkJoinPoll = new ForkJoinPool(64);

  public final String group;
  public final String app;
  private final AppConfig appConfig;
  private final EventLoopGroup eventLoopGroup;
  private final LoadBalance<Weightable> loadBalance;
  private final List<ClientFilter> clientFilters;

  /**
   * 成功的连接
   */
  private final ConcurrentMap<HostPort, NettyClientContext> activeConnectorMap = Maps
          .newConcurrentMap();
  /**
   * 失败的连接
   */
  private final ConcurrentMap<HostPort, NettyClientContext> zombieConnectorMap = Maps
          .newConcurrentMap();
  /**
   * methodId - MethodRouter
   */
  private final ConcurrentArrayList<MethodRouter> methodRouterMap = new ConcurrentArrayList<>();
  private final ConcurrentMap<String, Integer> methodStringToIdMap = Maps.newConcurrentMap();
  /**
   * methodId - ServiceMethodName
   */
  private final IntObjectHashMap<String> methodIdToServiceMethodNameMap = new IntObjectHashMap<>();
  /**
   * 服务端支持的服务接口
   */
  private final ConcurrentMap<String, Boolean> supportClassMap = Maps.newConcurrentMap();

  private volatile Thread rescueAndHeartBeatDeamonThread;

  private volatile long lastHeartBeatTime = 0;
  private volatile long lastRescueTime = 0;

  private volatile boolean closing = false;

  static {
    Runtime.getRuntime().addShutdownHook(new Thread(appForkJoinPoll::shutdown,
            "appForkJoinPoll-shutdown-thread"));
  }

  public App(AppConfig appConfig, EventLoopGroup eventLoopGroup,
             List<ClientFilter> clientFilters) {
    this.group = appConfig.getGroup();
    this.app = appConfig.getApp();
    this.appConfig = appConfig;
    this.eventLoopGroup = eventLoopGroup;
    this.loadBalance = appConfig.getLoadBalance();
    this.clientFilters = clientFilters;

    if (appConfig.getDiscover() != null) {
      appConfig.getDiscover().startRegistryListener(group, app, ProtocolType.rpc,
              serverWithWeight -> {
                if (closing) {
                  return;
                }

                if (log.isInfoEnabled()) {
                  log.info("discover检测到服务变化: " + serverWithWeight);
                }

                try {
                  setConnect(serverWithWeight);
                } catch (Exception e) {
                  log.error("添加新注册服务连接失败", e);
                }
              });
    }
  }

  public void setConnect(HostPort... hostPorts) throws Exception {
    if (ArrayUtil.isEmpty(hostPorts)) {
      return;
    }

    if (closing) {
      return;
    }

    Map<HostPort, Integer> collect = Arrays.stream(hostPorts)
            .collect(Collectors.toMap(t -> t, t -> HulkService.default_weight));
    setConnect(collect);
  }

  public void setConnect(Map<HostPort, Integer> serverWithWeight) throws Exception {
    if (MapUtil.isEmpty(serverWithWeight)) {
      return;
    }

    if (closing) {
      return;
    }

    // 后台线程任务，如心跳任务、连接修复任务等
    tryStartDeamonJob();

    AtomicBoolean changed = new AtomicBoolean(false);
    // 未建立连接的建立连接
    appForkJoinPoll.submit(() -> {
      serverWithWeight.entrySet()
              .parallelStream()
              .filter(kv -> !activeConnectorMap.containsKey(kv.getKey()))
              .forEach(kv -> {
                if (closing) {
                  return;
                }
                changed.set(true);
                setConnect(kv.getKey(), kv.getValue());
              });
    }).get();

    // 删除active中多余的连接
    appForkJoinPoll.submit(() -> {
      activeConnectorMap
              .entrySet()
              .parallelStream()
              .forEach(kv -> {
                if (closing) {
                  return;
                }

                HostPort serverAddress = kv.getKey();

                if (!serverWithWeight.containsKey(serverAddress)) {
                  changed.set(true);
                  NettyClientContext context = activeConnectorMap.remove(serverAddress);

                  try {
                    context.close();
                  } catch (IOException e) {
                    if (log.isWarnEnabled()) {
                      log.warn(serverAddress + "关闭失败", e);
                    }
                  }
                }
              });
    }).get();

    // 删除zombie中多余的连接
    appForkJoinPoll.submit(() -> {
      zombieConnectorMap
              .entrySet()
              .parallelStream()
              .forEach(kv -> {
                if (closing) {
                  return;
                }

                changed.set(true);
                HostPort serverAddress = kv.getKey();

                if (!serverWithWeight.containsKey(serverAddress)) {
                  NettyClientContext context = zombieConnectorMap.remove(serverAddress);

                  try {
                    context.close();
                  } catch (IOException e) {
                    if (log.isWarnEnabled()) {
                      log.warn(serverAddress + "关闭失败", e);
                    }
                  }
                }
              });
    }).get();

    // 重置方法路由
    if (changed.get()) {
      // 所有激活的连接
      Collection<NettyClientContext> connectors = activeConnectorMap.values();

      int length = methodRouterMap.size();
      for (int i = 0; i < length; i++) {
        MethodRouter router = methodRouterMap.get(i);
        router.setConnectors(connectors);
      }
    }
  }

  public boolean isSupport(Class<?> clazz) {
    return supportClassMap.containsKey(InvokerUtils.getServiceClassName(group, app, clazz));
  }

  /**
   * 获取methodId
   */
  public Integer getMethodId(Method method) {
    return getMethodId(InvokerUtils.getServiceMethodName(group, app, method));
  }

  private Integer getMethodId(String methodString) {
    Integer methodId = methodStringToIdMap.get(methodString);

    if (methodId != null) {
      return methodId;
    }

    synchronized (this) {
      methodId = methodStringToIdMap.get(methodString);

      if (methodId != null) {
        return methodId;
      }

      MethodRouter router = new MethodRouter(methodString, loadBalance);
      router.setConnectors(activeConnectorMap.values());

      methodRouterMap.add(router);

      methodId = methodRouterMap.size() - 1;

      methodStringToIdMap.put(methodString, methodId);
      methodIdToServiceMethodNameMap.put(methodId, methodString);
    }

    return methodId;
  }

  private synchronized void setConnect(HostPort serverAddress, int weight) {
    if (closing) {
      return;
    }

    NettyClientContext context = new NettyClientContext(eventLoopGroup, appConfig, clientFilters,
            serverAddress);

    try {
      context.connect();
      context.setWeight(weight);

      List<String> list = loadClass(context);
      for (String clazz : list) {
        supportClassMap.put(clazz, Boolean.TRUE);
      }

      if (log.isInfoEnabled()) {
        log.info(group + "#" + app + " " + serverAddress + " support services: " + list);
      }

      Map<String, Integer> methodStringToServiceIdMap = loadServiceId(context);
      context.setServiceMethodNameToServiceIdMap(methodStringToServiceIdMap);

      context.initSerializer();

      addConnect(context);
    } catch (Exception e) {
      if (log.isWarnEnabled()) {
        log.warn(serverAddress + "连接失败", e);
      }

      try {
        context.close();
      } catch (Exception e2) {
        // ignore
      }
    }
  }

  private synchronized void tryStartDeamonJob() {
    if (closing) {
      return;
    }

    if (null == rescueAndHeartBeatDeamonThread) {
      return;
    }

    Thread daemonThread = new Thread(() -> {
      while (!closing) {
        if (!closing && !appForkJoinPoll.isShutdown() &&
                System.currentTimeMillis() - lastRescueTime > rescue_period) {
          try {
            appForkJoinPoll.submit(this::rescue).get();
            lastRescueTime = System.currentTimeMillis();
          } catch (InterruptedException e) {
            break;
          } catch (ExecutionException e) {
            if (log.isWarnEnabled()) {
              log.warn(StringUtils.join(group, "#", app, " rescue error", e));
            }
          }
        }

        try {
          Thread.sleep(Math.min(rescue_period, heartbeat_period));
        } catch (InterruptedException e) {
          break;
        }

        if (!closing && !appForkJoinPoll.isShutdown() &&
                System.currentTimeMillis() - lastHeartBeatTime > heartbeat_period) {
          try {
            appForkJoinPoll.submit(this::heartbeat).get();
            lastHeartBeatTime = System.currentTimeMillis();
          } catch (InterruptedException e) {
            break;
          } catch (ExecutionException e) {
            if (log.isWarnEnabled()) {
              log.warn(StringUtils.join(group, "#", app, " heartbeat error", e));
            }
          }
        }
      }
    }, "app-heartbeat-and-rescue-thread-#" + group + "#" + app);
    daemonThread.setDaemon(true);
    daemonThread.start();
    this.rescueAndHeartBeatDeamonThread = daemonThread;
  }

  private void heartbeat() {
    // todo
  }

  private void rescue() {
    // todo
    if (closing) {
      return;
    }

    if (log.isDebugEnabled()) {
      log.debug(StringUtils.join(group, "#", app, " start rescue zombies"));
    }

    if (zombieConnectorMap.isEmpty()) {
      if (log.isDebugEnabled()) {
        log.debug(StringUtils.join(group, "#", app, " has no zombie connectors"));
      }
      return;
    }

    zombieConnectorMap.entrySet()
            .stream().parallel()
            .forEach(kv -> {
              if (closing) {
                return;
              }

              HostPort serverAddress = kv.getKey();
              NettyClientContext clientContext = kv.getValue();

              try {
                clientContext.connect();

                // 加载已经注册的服务接口
                List<String> list = loadClass(clientContext);
                for (String clazz : list) {
                  supportClassMap.put(clazz, true);
                }

                if (log.isInfoEnabled()) {
                  log.info(StringUtils.join("rescue ", group, "#", app, " "),
                          clientContext.getNettyClient().getServerAddress(), " support services: ", list);
                }

                Map<String, Integer> methodStringToServiceIdMap = loadServiceId(clientContext);

              } catch (Exception e) {
                if (log.isWarnEnabled()) {
                  log.warn(StringUtils.join("rescue ", group, "#", app, " ",
                          clientContext.getNettyClient().getServerAddress(), "fail"), e);
                }
              }
            });

  }

  /**
   * 获取方法全限定名与serviceId的映射
   */
  private Map<String, Integer> loadServiceId(NettyClientContext clientContext) throws Exception {
    int serviceId = HulkConnectService.SERVICE_METHOD_REGISTER;
    long timeout = HulkService.default_timeout;
    CompletableFuture<Map<String, Integer>> completableFuture = clientContext
            .execute(serviceId, timeout);

    return completableFuture.get();
  }

  /**
   * 获取服务端注册的服务（接口全限定名称）
   */
  private List<String> loadClass(NettyClientContext clientContext)
          throws ExecutionException, InterruptedException {
    int serviceId = HulkConnectService.SERVICE_CLASS_REGISTER;
    long timeout = HulkService.default_timeout;
    CompletableFuture<List<String>> completableFuture = clientContext.execute(serviceId, timeout);

    return completableFuture.get();
  }

  private synchronized void addConnect(NettyClientContext context) throws Exception {
    if (closing) {
      return;
    }

    if (context == null) {
      return;
    }

    HostPort serverAddress = context.getNettyClient().getServerAddress();
    activeConnectorMap.put(serverAddress, context);
    zombieConnectorMap.remove(serverAddress);
  }

  @Override
  public void close() throws IOException {
    if (closing) {
      return;
    }

    closing = true;
    rescueAndHeartBeatDeamonThread.interrupt();

    activeConnectorMap.forEach((key, connectorContext) -> {
      try {
        connectorContext.close();
      } catch (Exception e) {
        if (log.isWarnEnabled()) {
          log.warn(group + "#" + app + " " + "关闭过程中发生错误", e);
        }
      }
    });

    activeConnectorMap.clear();
    zombieConnectorMap.clear();
    methodStringToIdMap.clear();
    methodIdToServiceMethodNameMap.clear();
    supportClassMap.clear();

    if (appConfig.getDiscover() != null) {
      appConfig.getDiscover().close();
    }
  }

  public CompletableFuture<?> execute(int methodId, long timeout, MethodParam methodParam,
                                      Invoker<CompletableFuture<?>> failoverInvoker) {
    MethodRouter methodRouter;
    if (methodId < 0 || methodId > methodRouterMap.size()
            || (methodRouter = methodRouterMap.get(methodId)) == null) {
      String errMsg = StringUtils.join(group, "#", app, "不支持方法id:", methodId);
      log.error(errMsg);
      return CompletableFuture.failedFuture(new RemoteException(errMsg));
    }

    // 有filter的话可能会需要在onRequest或onResponse的时候用到
    if (clientFilters.size() > 0) {
      RemoteContext.setRemoteMethod(methodRouter.getMethod());
      RemoteContext.setServiceMethodName(methodRouter.getServiceMethodName());
    }

    NettyClientContext nettyClientContext = methodRouter.selectConnector();
    for (int i = 0; i < max_connector_select_times; i++) {
      if (nettyClientContext == null) {
        break;
      }

      if (nettyClientContext.isClosed()) {
        try {
          nettyClientContext.close();
        } catch (Exception e) {
          // ignore
        }

        nettyClientContext = methodRouter.selectConnector();
      } else if (nettyClientContext.isZombie()) {
        kill(nettyClientContext);
        nettyClientContext = methodRouter.selectConnector();
      } else {
        break;
      }
    }

    if (nettyClientContext == null) {
      String errMsg = StringUtils.join(group, "#", app, "无可用的客户端连接");

      if (log.isWarnEnabled()) {
        log.warn(errMsg);
      }

      if (failoverInvoker == null) {
        return CompletableFuture.failedFuture(new RemoteException(errMsg, false));
      } else {
        return failoverInvoker.invoke(methodParam);
      }
    }

    int serviceId = getServiceId(nettyClientContext, methodId);
    if (serviceId < 0) {
      String errMsg = StringUtils.join(group, "#", app, "找不到对应的服务, methodId为", methodId);
      if (log.isWarnEnabled()) {
        log.warn(errMsg);
      }

      return CompletableFuture.failedFuture(new RemoteException(errMsg, false));
    }

    return nettyClientContext.execute(serviceId, timeout, methodParam, failoverInvoker);
  }

  /**
   * 获取远程服务id
   */
  private int getServiceId(NettyClientContext nettyClientContext, int methodId) {
    int serviceId = nettyClientContext.getServiceId(methodId);
    if (serviceId < 0) {
      String serviceMethodName = methodIdToServiceMethodNameMap.get(methodId);

      if (serviceMethodName == null) {
        String msg = group + "#" + app + " " + "找不到对应的服务, methodId: " + methodId;

        if (log.isWarnEnabled()) {
          log.warn(msg);
        }

        return -1;
      }

      nettyClientContext.putServiceId(serviceMethodName, methodId);
      serviceId = nettyClientContext.getServiceId(methodId);

      if (serviceId < 0) {
        String msg = group + "#" + app + " " + "找不到对应的服务, methodId: " + methodId + ", method: "
                + serviceMethodName;

        if (log.isWarnEnabled()) {
          log.warn(msg);
        }

        return -1;
      }
    }

    return 0;
  }

  /**
   * kill有问题的连接
   * <p>
   * 因为连接是池化资源，所以会存在多线程问题
   */
  private synchronized void kill(NettyClientContext nettyClientContext) {
    if (nettyClientContext == null) {
      return;
    }

    HostPort serverAddress = nettyClientContext.serverAddress;

    if (log.isWarnEnabled()) {
      log.warn(group + "#" + app + " " + serverAddress + " is zombie, and will be killed!");
    }

    activeConnectorMap.remove(serverAddress);
    zombieConnectorMap.put(serverAddress, nettyClientContext);

    // 重新构建该context关联的路由信息
    Collection<NettyClientContext> connectors = activeConnectorMap.values();
    int length = methodRouterMap.size();
    for (int i = 0; i < length; i++) {
      MethodRouter router = methodRouterMap.get(i);

      if (!nettyClientContext.isSupport(router.getServiceMethodName())) {
        continue;
      }

      router.setConnectors(connectors);
    }

    if (log.isWarnEnabled()) {
      log.warn(group + "#" + app + " " + serverAddress + " is zombie, and have been killed!");
    }
  }
}
